package ba.keco.hbase.lib;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

public class HBaseShowcaseInputFormat extends InputFormat<NullWritable, NullWritable> {

	public class HBaseShowcaseRecordReader extends RecordReader<NullWritable, NullWritable> {

		private int numberOfRecords;

		private int counter = 0;

		@Override
		public void close() throws IOException {
		}

		@Override
		public NullWritable getCurrentKey() throws IOException, InterruptedException {
			return NullWritable.get();
		}

		@Override
		public NullWritable getCurrentValue() throws IOException, InterruptedException {
			return NullWritable.get();
		}

		@Override
		public float getProgress() throws IOException, InterruptedException {
			if (counter == 0){
				return 0;
			}
			return numberOfRecords/counter;
		}

		@Override
		public void initialize(InputSplit inputSplit, TaskAttemptContext context) throws IOException, InterruptedException {
			numberOfRecords = context.getConfiguration().getInt("hbase.showcase.input.records", 10);
		}

		@Override
		public boolean nextKeyValue() throws IOException, InterruptedException {
			counter++;
			return counter <= numberOfRecords;
		}

	}
	
	@Override
	public RecordReader<NullWritable, NullWritable> createRecordReader(InputSplit arg0, TaskAttemptContext arg1) throws IOException, InterruptedException {
		return new HBaseShowcaseRecordReader();
	}

	@Override
	public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException {
		int mappers = context.getConfiguration().getInt("hbase.showcase.mappers", 1);
		List<InputSplit> splits = new ArrayList<InputSplit>(mappers);
		for (int i = 0; i < mappers; i++) {
			splits.add(new FileSplit(new Path("/"), 0, 10, new String[]{"test"}));
		}
		return splits;
	}

}
